package com.kafka.xiang.kakfamodual;

public class KafkaListener {
    private Thread thread;
    private String server;
    private String listenTopic;
    private String groupId;
    private Runnable listeningFun;
    private Runnable terminateFun;//Runnable 接口是 Java 中表示可运行任务的接口,因为它只有一个抽象方法 run()
    private boolean consumerActive;
    private boolean listenerActive;

    public KafkaListener(String server, String listenTopic, String groupId, Runnable listeningFun, Runnable terminateFun) {
        this.server = server;
        this.listenTopic = listenTopic;
        this.groupId = groupId;
        this.listeningFun = listeningFun;
        this.terminateFun = terminateFun;
        this.consumerActive = false;
        this.listenerActive = false;
    }

    public boolean startListening() {
        consumerActive = true;
        thread = new Thread(listeningFun);//java的Thread不用把函数名称和参数分开
        thread.start();
        listenerActive = true;
        return listenerActive;
    }

    public void stopListening() {
        terminateFun.run();
        try {//线程阻塞
            thread.join();//Thread.join() 是一个方法，用于等待thread线程的结束。当调用 join() 方法时，当前线程将阻塞（暂停执行），直到被调用 join() 方法的线程执行完成。即thread执行完毕
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        listenerActive = false;
        consumerActive = false;
    }
}
